package com.example.demo.producer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
@Component
public class ProducerTest {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @PostConstruct
    private void init() {
        log.info("ProducerTest 开始初始化...");
        ExecutorService executorService = Executors.newFixedThreadPool(10);

        for (int i = 1; i <= 100; i++) {
            int finalI = i;
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    log.info("start send {}", finalI);
                    kafkaTemplate.send("TOPIC_INSERT", String.valueOf(finalI));
                    log.info("end send {}", finalI);
                }
            });
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    log.info("start send {}", finalI);
                    kafkaTemplate.send("TOPIC_UPDATE", String.valueOf(finalI));
                    log.info("end send {}", finalI);
                }
            });
        }

        log.info("ProducerTest 完成初始化");
    }

}
